package main

import (
	"fmt"
	"github.com/Shopify/sarama"
	"time"
)

// 点对点，发布订阅
// 广播推送，单个推送，群组推送

// 生产者
func main() {
	PushMsg()

}

// 广播推送
func PushMsg() {
	// 创建一个config
	addr := []string{"192.168.72.133:9092"}
	config := sarama.NewConfig()
	config.Version = sarama.V2_1_0_0
	config.Producer.RequiredAcks = sarama.WaitForAll
	config.Producer.Partitioner = sarama.NewRandomPartitioner
	config.Producer.Return.Successes = true // 同步模式
	config.Producer.Return.Errors = true
	//admin, err := sarama.NewClusterAdmin(addr, config)
	//if err != nil {
	//	fmt.Println(err)
	//}
	//err = admin.CreateTopic("test", &sarama.TopicDetail{NumPartitions: 1, ReplicationFactor: 3}, false)
	//if err != nil {
	//	fmt.Println("创建topic失败：", err)
	//}
	//
	//err = admin.Close()
	//if err != nil {
	//	fmt.Println(err)
	//}

	// 创建一个消费者
	producer, err := sarama.NewSyncProducer(addr, config)
	if err != nil {
		fmt.Println(err)
	}

	defer func() {
		if err := producer.Close(); err != nil {
			fmt.Println(err)
		}
	}()

	// 创建一条消息
	msg := &sarama.ProducerMessage{Topic: "test", Value: sarama.StringEncoder("testing 123")}
	a := 0
	for {
		// 发送消息
		partition, offset, err := producer.SendMessage(msg)
		if err != nil {
			fmt.Println("failed to send message: ", err)
		} else {
			fmt.Printf("message sent to partition %d at offset %d\n", partition, offset)
		}
		time.Sleep(time.Second * 2)
		a++
		if a > 2 {
			break
		}
	}

}
